package com.esiag.isidis.fi.pdsTeam.jms.consumer;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import com.esiag.isidis.fi.pdsTeam.jms.connection.AbstractConnectionFactory;
import com.esiag.isidis.fi.pdsTeam.send.beans.ScalabilityMessage;
/* classe pour adapter activeMqConsumer pour que notre systeme soit scalable
/*
 * premiere class a recevoir le message  
 */
public class ScalabilityConsumer implements MessageListener {

  private AbstractConsumerFactory<?> mAbstractConsumerFactory;
  private AbstractConnectionFactory mConnectionFactory;
  private ScalabilityChargeConsumer mScalabilityChargeConsumer;

  public ScalabilityConsumer(AbstractConsumerFactory<?> pAbstractConsumerFactory) {
    mAbstractConsumerFactory = pAbstractConsumerFactory;
    try {
      mConnectionFactory = mAbstractConsumerFactory.getConnection(ScalabilityMessage.class);
      mConnectionFactory.start();
      MessageConsumer consumer = mConnectionFactory.createTopicConsumer(mAbstractConsumerFactory
          .getQueue(ScalabilityMessage.class));
      mScalabilityChargeConsumer = new ScalabilityChargeConsumer();
      consumer.setMessageListener(mScalabilityChargeConsumer);
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
//notifie les autres que la charge a augmentée 
  public void sendCharge(Integer pCharge) {
    try {
      ObjectMessage msg;
      Session session = mConnectionFactory.getSession();
      msg = session.createObjectMessage();
      ScalabilityMessage sm = new ScalabilityMessage();
      sm.setCharge(pCharge);
      msg.setObject(sm);
      MessageProducer producer = mConnectionFactory.createTopicProducer(mAbstractConsumerFactory
          .getQueue(ScalabilityMessage.class));
      producer.send(msg);
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }

  public void notifyAdmin() {
	  System.out.println("il faut lancer une nouvelle machine le system est overloaded certaine seront traite plutard");
  }
  
  public void onOverLoadedEvent(){
	  new Thread(new Runnable() {
		public void run() {
			// on attend jusqu a ce que le system sois normal
			while(mScalabilityChargeConsumer.isOverLoaded()){
				try{
					Thread.sleep(1000);
				}catch(Exception ex){
					ex.printStackTrace();
				}
			}
			// on essaie a nouveau de traite le message qui est au debut de la pile
			onMessage(mScalabilityChargeConsumer.retrieveObject());
		}
	}).start();
  }
  
  public void onMessageThread(final Message pMessage){
	  new Thread(new Runnable() {
		public void run() {
				synchronized (mAbstractConsumerFactory) {
				    if (mScalabilityChargeConsumer.isOverLoaded()) {
				    	notifyAdmin();
				    	// les message qu on peux pas traite on les met dans une file pour les traite plutard
				    	// quand le system serra plus leger
				    	mScalabilityChargeConsumer.addObject(pMessage);
				    	onOverLoadedEvent();
				    	return;
				    }	
				    sendCharge(1);
				}
			    mAbstractConsumerFactory.onMessage(pMessage);
			    sendCharge(-1);
		}
	}).start();
  }

  public void onMessage(Message pMessage) {
	  onMessageThread(pMessage);
  }
}
